package com.xhg.mqtt.core;

import com.google.protobuf.ByteString;
import com.xhg.mqtt.constant.MqttConstants;
import com.xhg.mqtt.constant.MsgConstants;
import com.xhg.mqtt.dto.EventCodeEnum;
import com.xhg.mqtt.dto.MqttConfigInfoDTO;
import com.xhg.mqtt.job.MessageRepeatQueue;
import com.xhg.mqtt.listener.MqttLoggerFactory;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttException;

/**
 * mqtt操作客户端（主要是给app使用的） 发送消息，订阅，ack等
 *
 * @author E.T
 * @version 1.0
 * @Description
 * @date 2018-07-17 19:40
 * @updateby
 * @updatedate
 * @since 1.0
 */
public class MqttSimpleClient {

    private MqttMessageHandler messageHandler;
    private String deviceId;
    private String version;
    private MqttSimpleClient() {
    }
    private static class MqttSimpleClientHolder {
        private final static MqttSimpleClient instance = new MqttSimpleClient();
    }
    public static MqttSimpleClient getInstance() {
        return MqttSimpleClientHolder.instance;
    }

    {
        /**
         * 初始化一些监控
         */
        MessageRepeatQueue.getInstance().init();
    }

    /**
     * 初始化信息
     *
     * @param mqttInfo mqtt信息
     * @param deviceId 设备id
     * @param version 版本号
     */
    public  void init(MqttConfigInfoDTO mqttInfo, String deviceId,
        String version){
        this.version = version;
        this.deviceId = deviceId;
        messageHandler = new MqttMessageHandler(mqttInfo);
        if (isConnect()) {
            return;
        }
        //真正初始化mqtt连接
        MqttClient.getInstance().connection(mqttInfo, deviceId, version);
        //连接完成上报设备信息
        connectComplete();
    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param topic 主题
     * @param pushMessage 消息内容
     */
    public void publish(String topic, int qos, byte[] pushMessage) {
        publish(qos, MqttConstants.RETAINED_FALSE, topic, pushMessage);
    }

    /**
     * 发布消息
     *
     * @param qos 发布质量 0 1 2
     * @param retained 是否持久化消息
     * @param topic 主题
     * @param pushMessage 消息内容
     */
    public void publish(int qos, boolean retained, String topic, byte[] pushMessage) {
        try {
            MqttClient.getMqttInstance().publish(topic, pushMessage, qos, retained);
        } catch (Exception e) {
            MqttLoggerFactory.error(MsgConstants.SEND_ERROR ,e);
        }
    }


    /**
     * 订阅某个主题，qos默认为0 回调到统一的callback
     *
     * @param topic 主题
     */
    public void subscribe(String topic) {
        subscribe(topic, MqttConstants.QOS);
    }

    /**
     * 订阅某个主题 回调到统一的callback
     *
     * @param topic 主题
     * @param qos 消息质量 0 1 2
     */
    public void subscribe(String topic, int qos) {
        try {
            MqttClient.getMqttInstance().subscribe(topic, qos);
        } catch (MqttException e) {
            MqttLoggerFactory.error(MsgConstants.SUB_ERROR, e);
        }
    }

    /**
     * 取消订阅某个主题
     *
     * @param topic 主题
     */
    public void unsubscribe(String topic) {
        try {
            MqttClient.getMqttInstance().unsubscribe(topic);
        } catch (MqttException e) {
            MqttLoggerFactory.error(MsgConstants.SUB_ERROR, e);
        }
    }

    /**
     * 订阅某个topic，有独立的callback
     *
     * @param topic 主题
     * @param listener 消息监听器
     */
    public void subscribe(String topic, IMqttMessageListener listener) {
        try {
            MqttClient.getMqttInstance().subscribe(topic, listener);
        } catch (MqttException e) {
            MqttLoggerFactory.error(MsgConstants.SUB_ERROR, e);
        }
    }

    /**
     * 订阅多个topic
     *
     * @param topic 主题
     * @param qos 消息质量 0 1 2
     */
    public void subscribe(String[] topic, int[] qos) {
        try {
            MqttClient.getMqttInstance().subscribe(topic, qos);
        } catch (MqttException e) {
            MqttLoggerFactory.error(MsgConstants.SUB_ERROR, e);
        }
    }

    /**
     * 取消多个订阅某个主题
     *
     * @param topics 多个主题
     */
    public void unsubscribe(String[] topics) {
        try {
            MqttClient.getMqttInstance().unsubscribe(topics);
        } catch (MqttException e) {
            MqttLoggerFactory.error(MsgConstants.SUB_ERROR, e);
        }
    }

    /**
     * 断开链接
     */
    private void disconnect() {
        try {
            MqttClient.getMqttInstance().disconnect();
        } catch (MqttException e) {
            MqttLoggerFactory.error(MsgConstants.BREAK_ERROR, e);
        }
    }


    /**
     * 是否连接
     */
    public boolean isConnect() {
        return MqttClient.getMqttInstance() == null ? false
            : MqttClient.getMqttInstance().isConnected();
    }

    /**
     * 发送消息
     *
     * @param eventCodeEnum 指令码
     * @param deviceId 设备id
     * @param version 版本号
     * @param cc 控制码 0不需要回复 1需要回复
     * @param body 具体业务消息体
     */
    public boolean pushMessage(EventCodeEnum eventCodeEnum, String deviceId, String version, int cc,
        int qos,
        ByteString body)
        throws Exception {
        return messageHandler.handleMessage(eventCodeEnum, deviceId, version, cc, qos, body);
    }

    /**
     * 发送ack消息
     *
     * @param eventCodeEnum 指令码
     * @param messageId 消息id 跟接收到的消息id一致
     * @param version 版本号
     */

    public boolean ackMessage(EventCodeEnum eventCodeEnum, String messageId, String deviceId,
        String version, int qos)
        throws Exception {
        return messageHandler.ackMessage(eventCodeEnum, messageId, deviceId, version, qos);
    }


    /**
     * 连接初始化成功
     */
    public void connectComplete() {
        if (!isConnect()) {
            return;
        }
        messageHandler.login(deviceId, version);
    }
}